package com.flink.batch.core;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;


/**
 * @version v1.0
 * @ProjectName flinkwarehouse
 * @ClassName BatchWordCountDemo
 * @Description TODO 批处理的实例
 * @Author wudl
 * @Date 2020/8/28 11:25
 */

public class BatchWordCountDemo {
    public static void main(String[] args) throws Exception {
        // 1.获取批处理的执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<String> ds = env.fromElements("hdfs hdfs hive", "spark flink flink");

        // 读取输入的数据
        DataSet<Tuple2<String, Integer>> sum = ds.flatMap(new LineSplitter()).groupBy(0).sum(1);

        sum.print();

    }


    public static class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {

        @Override
        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
            for (String world : s.split(" ")) {
                collector.collect(new Tuple2<>(world, 1));
            }
        }
    }


}
